﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Algobox.Objects.Finance.Prices;
using Algobox.Structure.Finance.Prices;
using Algobox.Objects.Finance.Assets.Stock;
using Algobox.Feeds.Finance.MarketData;
using Algobox.Toolbox.Collections;
using Algobox.Objects.Finance;


namespace Algobox.Apps.Finance.SterlingTrader
{

    public class QuoteAdaptor : Algobox.Interop.SterlingTrader.QuoteManagement, IMarketFeedAdaptor
    {
        #region Constructors

        public QuoteAdaptor(string feedName)
        {
            FeedName = feedName;
        }

        #endregion
        //////////
        #region Member Variables

        /// <summary>
        /// Logger used by this class
        /// </summary>        
        private static readonly NLog.Logger Log = NLog.LogManager.GetCurrentClassLogger();

        //ConcurrentDictionary<string, MarketFeedRequest> _mapRequests = new ConcurrentDictionary<string, MarketFeedRequest>();

        //public event Action<IStock, PriceVolume, PriceVolume> OnPriceBBO;
        //public event Action<IStock, PriceVolume, PriceVolume> OnPriceBBOAuction;
        //public event Action<IStock, PriceVolume> OnPriceAsk;
        //public event Action<IStock, PriceVolume> OnPriceBid;
        //public event Action<IStock, PriceVolume> OnPriceLast;
        //public event Action<IStock> OnPhase;

        ConcurrentDictionary<string, StockSecurity> _mapSecurities = new ConcurrentDictionary<string, StockSecurity>();
        ConcurrentDictionary<string, RequestCounter> _mapSecuritiesCount = new ConcurrentDictionary<string, RequestCounter>();

        ConcurrentDictionary<IStock, List<IMarketUpdateAsk>> _mapAskRequests = new ConcurrentDictionary<IStock, List<IMarketUpdateAsk>>();
        ConcurrentDictionary<IStock, List<IMarketUpdateBid>> _mapBidRequests = new ConcurrentDictionary<IStock, List<IMarketUpdateBid>>();
        ConcurrentDictionary<IStock, List<IMarketUpdateLast>> _mapLastRequests = new ConcurrentDictionary<IStock, List<IMarketUpdateLast>>();
        ConcurrentDictionary<IStock, List<IMarketUpdatePhase>> _mapPhaseRequests = new ConcurrentDictionary<IStock, List<IMarketUpdatePhase>>();
        ConcurrentDictionary<IStock, List<IMarketUpdateOpen>> _mapOpenRequests = new ConcurrentDictionary<IStock, List<IMarketUpdateOpen>>();
        ConcurrentDictionary<IStock, int> _mapRawRequests = new ConcurrentDictionary<IStock, int>();


        public event Action<IStock, long> OnSubscriptionAdded;
        public event Action<IStock, long> OnSubscriptionRemoved;
        public event Action OnMarketFeedShutdown;
        
        /// <summary>
        /// Events for raw access to Sterling Quote updates
        /// Action<TIMESTAMP,SYBMOL,PHASE,LAST,LASTSIZE,ISLAST,BID,BIDSIZE,ISBID,ASK,ASKSIZE,ISASK> 
        /// </summary>
        public event Action<DateTime, string, Phases.Phase, double, int, bool, double, int, bool, double, int, bool> OnRawQuoteUpdate;
   
        private class RequestCounter
        {
            long _count;

            public long Count { get { return _count; } }

            public long Increment()
            {
                return System.Threading.Interlocked.Increment(ref _count);
            }

            public long Decrement()
            {
                return System.Threading.Interlocked.Decrement(ref _count);
            }
        }

        #endregion
        //////////
        #region Properties
        
        public int NumberOfSecurityRequests { get { return _mapSecurities.Count; } }

        #endregion
        //////////
        #region Functions

        public override void PostProcessSterlingShutdown()
        {
            if (OnMarketFeedShutdown != null)
                OnMarketFeedShutdown();
        }

        public override void PostProcessQuoteSnapshot(ref SterlingLib.structSTIQuoteSnap stiQuoteSnap)
        {
            StockSecurity security;

            DateTime timeStamp = DateTime.Now;

            if (_mapSecurities.TryGetValue(stiQuoteSnap.bstrSymbol, out security))
            {
                if (stiQuoteSnap.bAskPrice == 1)
                {
                    double price = stiQuoteSnap.fAskPrice;
                    ulong volume = (ulong)stiQuoteSnap.nAskSize;

                    /*
                     * Don't send updates on snapshot
                    _mapAskRequests.AddOrUpdate(security, new List<IMarketUpdateAsk>(),
                        (key, list) =>
                        {
                            list.ForEach(callback => callback.OnPriceAsk(key, new PriceVolume(price, volume)));
                            return list;
                        });
                    */

                    // update security with current prices
                    security.Prices.Ask.Update(price, volume, timeStamp);
                }

                if (stiQuoteSnap.bBidPrice == 1)
                {
                    double price = stiQuoteSnap.fBidPrice;
                    ulong volume = (ulong)stiQuoteSnap.nBidSize;

                    /*
                     * Don't send updates on snapshot
                    _mapBidRequests.AddOrUpdate(security, new List<IMarketUpdateBid>(),
                        (key, list) =>
                        {
                            list.ForEach(callback => callback.OnPriceBid(key, new PriceVolume(price, volume)));
                            return list;
                        });
                     */

                    // update security with current prices
                    security.Prices.Bid.Update(price, volume, timeStamp);
                }

                if (stiQuoteSnap.bLastPrice == 1)
                {
                    double price = stiQuoteSnap.fLastPrice;
                    ulong volume = (ulong)stiQuoteSnap.nLastSize;

                    /*
                     * Don't send updates on snapshot
                    _mapLastRequests.AddOrUpdate(security, new List<IMarketUpdateLast>(),
                        (key, list) =>
                        {
                            list.ForEach(callback => callback.OnPriceLast(key, new PriceVolume(price, volume)));
                            return list;
                        });
                    */

                    security.Prices.Last.Price = price;
                    security.Prices.Last.Volume = volume;
                    security.Prices.OHLC.Volume = (ulong)stiQuoteSnap.nCumVolume;
                    security.Prices.VWAP = stiQuoteSnap.fVwap;
                }

                if (stiQuoteSnap.bHighPrice == 1)
                {
                    security.Prices.OHLC.High = stiQuoteSnap.fHighPrice;
                }
                if (stiQuoteSnap.bLowPrice == 1)
                {
                    security.Prices.OHLC.Low = stiQuoteSnap.fLowPrice;
                }
                if (stiQuoteSnap.bOpenPrice == 1)
                {
                    security.Prices.OHLC.Close = stiQuoteSnap.fClosePrice;
                    security.Prices.OHLC.Open = stiQuoteSnap.fOpenPrice;
                }
            }
            else
            {
                Log.Warn("No request mapped to symbol " + stiQuoteSnap.bstrSymbol);
            }
        }


        public override void PostProcessQuoteUpdate(ref SterlingLib.structSTIQuoteUpdate stiQuoteUpdate)
        {
            if (stiQuoteUpdate.bstrHaltResumeStatus != null)
                Log.Warn("bstrHaltResumeStatus = " + stiQuoteUpdate.bstrHaltResumeStatus);

            StockSecurity security;
            //DateTime timeStamp = DateTime.ParseExact(stiQuoteUpdate.bstrUpdateTime, "yyyyMMddHHmmss", System.Globalization.CultureInfo.InvariantCulture);
            
            DateTime timeStamp = Algobox.Toolbox.Diagnostics.DateTimePrecise.Now;

            bool isUpdate = false;

            if (_mapSecurities.TryGetValue(stiQuoteUpdate.bstrSymbol, out security))
            {
                // for now lets only look at bid and ask if both are set
                // otherwise there is no other way to detect auctions
                if (stiQuoteUpdate.bAskPrice == 1 && stiQuoteUpdate.bBidPrice == 1)
                {
                    // check we have a valid price
                    if (stiQuoteUpdate.bstrBidExch == stiQuoteUpdate.bstrAskExch
                        && stiQuoteUpdate.fBidPrice >= stiQuoteUpdate.fAskPrice)
                    {
                        // detected auction
                        if (security.Phase != Objects.Finance.Phases.Phase.IntradayAuction)
                        {
                            Log.Debug("Auction phase detected in {0} with {1}<->{2}", stiQuoteUpdate.bstrSymbol, stiQuoteUpdate.fBidPrice, stiQuoteUpdate.fAskPrice);
                            security.Phase = Objects.Finance.Phases.Phase.IntradayAuction;
                            _mapPhaseRequests.AddOrUpdate(security, new List<IMarketUpdatePhase>(),
                                    (key, list) =>
                                    {
                                        list.ForEach(callback => callback.OnPhaseChange(security));
                                        return list;
                                    });
                        }
                    }
                    else
                    {
                        // regular trading 
                        if (security.Phase != Objects.Finance.Phases.Phase.Open)
                        {
                            Log.Debug("Trading phase detected in {0} with {1}<->{2}", stiQuoteUpdate.bstrSymbol, stiQuoteUpdate.fBidPrice, stiQuoteUpdate.fAskPrice);
                            security.Phase = Objects.Finance.Phases.Phase.Open;
                            _mapPhaseRequests.AddOrUpdate(security, new List<IMarketUpdatePhase>(),
                                    (key, list) =>
                                    {
                                        list.ForEach(callback => callback.OnPhaseChange(security));
                                        return list;
                                    });
                        }

                        PriceVolume ask = null;
                        PriceVolume bid = null;

                        // ask price updated
                        if (stiQuoteUpdate.fAskPrice != 0 && stiQuoteUpdate.nAskSize != 0)
                        {
                            isUpdate = true;

                            ask = new PriceVolume(PriceVolume.PriceType.Ask, stiQuoteUpdate.fAskPrice, (ulong)stiQuoteUpdate.nAskSize, timeStamp);

                            _mapAskRequests.AddOrUpdate(security, new List<IMarketUpdateAsk>(),
                                (key, list) =>
                                {
                                    list.ForEach(callback => callback.OnPriceAsk(key, ask));
                                    return list;
                                });

                            // update security with current prices
                            security.Prices.Ask.Update(stiQuoteUpdate.fAskPrice, ask.Volume, timeStamp);
                        }

                        // bid price updated
                        if (stiQuoteUpdate.fBidPrice != 0 && stiQuoteUpdate.nBidSize != 0)
                        {
                            isUpdate = true;

                            bid = new PriceVolume(PriceVolume.PriceType.Bid, stiQuoteUpdate.fBidPrice, (ulong)stiQuoteUpdate.nBidSize, timeStamp);

                            _mapBidRequests.AddOrUpdate(security, new List<IMarketUpdateBid>(),
                                (key, list) =>
                                {
                                    list.ForEach(callback => callback.OnPriceBid(key, bid));
                                    return list;
                                });

                            // update security with current prices
                            security.Prices.Bid.Update(stiQuoteUpdate.fBidPrice, bid.Volume, timeStamp);
                        }
                    }
                }

                if (stiQuoteUpdate.bLastPrice == 1 && stiQuoteUpdate.fLastPrice != 0 && stiQuoteUpdate.nLastSize != 0)
                {
                    isUpdate = true;

                    security.Prices.VWAP = stiQuoteUpdate.fVwap;

                    PriceVolume last = new PriceVolume(PriceVolume.PriceType.Last, stiQuoteUpdate.fLastPrice, (ulong)stiQuoteUpdate.nLastSize, timeStamp);
                    _mapLastRequests.AddOrUpdate(security, new List<IMarketUpdateLast>(),
                            (key, list) =>
                            {
                                list.ForEach(callback => callback.OnPriceLast(key, last));
                                return list;
                            });

                    // update security with current prices
                    security.Prices.OHLC.Update(stiQuoteUpdate.fLastPrice, last.Volume, timeStamp);
                    security.Prices.Last.Update(stiQuoteUpdate.fLastPrice, last.Volume, timeStamp);
                }


                if (stiQuoteUpdate.bOpenPrice == 1)
                {                    
                    double openPrice = stiQuoteUpdate.fOpenPrice;
                    int openInterest = stiQuoteUpdate.nOpenInterest;
                    _mapOpenRequests.AddOrUpdate(security, new List<IMarketUpdateOpen>(),
                        (key, list) =>
                        {
                            list.ForEach(callback => callback.OnPriceOpen(key, openPrice, openInterest));
                            return list;
                        });

                    security.Prices.OHLC.Open = stiQuoteUpdate.fOpenPrice;                            
                }
                

                if (OnRawQuoteUpdate != null && isUpdate)
                {
                    OnRawQuoteUpdate(timeStamp,
                        stiQuoteUpdate.bstrSymbol,
                        security.Phase,
                        stiQuoteUpdate.fLastPrice,
                        stiQuoteUpdate.nLastSize,
                        stiQuoteUpdate.bLastPrice == 1,
                        stiQuoteUpdate.fBidPrice,
                        stiQuoteUpdate.nBidSize,
                        stiQuoteUpdate.bBidPrice == 1,
                        stiQuoteUpdate.fAskPrice,
                        stiQuoteUpdate.nAskSize,
                        stiQuoteUpdate.bAskPrice == 1);
                }
            }
            else
            {
                Log.Warn("No request mapped to symbol " + stiQuoteUpdate.bstrSymbol);
            }
        }


        #region IMarketFeedAdaptor Members

        public string FeedName { get; private set; }

        public bool IsStarted { get; private set; }

        public override bool Start()
        {
            if (!IsStarted)
            {
                try
                {
                    if (base.Start())
                    {
                        foreach (KeyValuePair<string, StockSecurity> stock in _mapSecurities)
                        {
                            if (_mapSecuritiesCount[stock.Key].Count > 0)
                                RegisterQuote(stock.Value);
                        }
                        IsStarted = true;
                    }
                }
                catch
                {
                    return false;
                }
            }
            return IsStarted;
        }

        public override void Stop()
        {
            if (IsStarted)
            {
                IsStarted = false;
                base.Stop();
            }
        }

        public void SubscribeBid(IStock stock, IMarketUpdateBid callback)
        {
            long requestCount = 0;

            _mapBidRequests.AddOrUpdate(stock,
                (key) =>
                {
                    requestCount = _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                        (symbol) =>
                        {
                            RequestCounter value = new RequestCounter();
                            value.Increment();
                            return value;
                        },
                        (symbol, value) =>
                        {
                            value.Increment();
                            return value;
                        })
                        .Count;

                    return new List<IMarketUpdateBid>() { callback };
                },
                (key, list) =>
                {
                    if (!list.Contains(callback))
                    {
                        list.Add(callback);

                        requestCount = _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                            (symbol) =>
                            {
                                RequestCounter value = new RequestCounter();
                                value.Increment();
                                return value;
                            },
                            (symbol, value) =>
                            {
                                value.Increment();
                                return value;
                            })
                            .Count;
                    }
                    else { } // don't subscribe twice you twat!
                    return list;
                }
            );

            if (OnSubscriptionAdded != null)
            {
                OnSubscriptionAdded(stock, requestCount);
            }

            if (requestCount == 1)
            {
                if (IsStarted)
                    RegisterQuote(stock);

                if (!_mapSecurities.ContainsKey(stock.Symbol))
                {
                    if (stock is StockSecurity)
                        _mapSecurities.TryAdd(stock.Symbol, (StockSecurity)stock);
                    else if (stock is StockListing)
                        _mapSecurities.TryAdd(stock.Symbol, ((StockListing)stock).Security);
                }
            }
        }

        public void SubscribeAsk(IStock stock, IMarketUpdateAsk callback)
        {
            long requestCount = 0;
            _mapAskRequests.AddOrUpdate(stock,
                (key) =>
                {
                    requestCount = _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                        (symbol) =>
                        {
                            RequestCounter value = new RequestCounter();
                            value.Increment();
                            return value;
                        },
                        (symbol, value) =>
                        {
                            value.Increment();
                            return value;
                        })
                        .Count;

                    return new List<IMarketUpdateAsk>() { callback };
                },
                (key, list) =>
                {
                    if (!list.Contains(callback))
                    {
                        list.Add(callback);

                        requestCount = _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                        (symbol) =>
                        {
                            RequestCounter value = new RequestCounter();
                            value.Increment();
                            return value;
                        },
                        (symbol, value) =>
                        {
                            value.Increment();
                            return value;
                        })
                        .Count;
                    }
                    else { } // don't subscribe twice you twat!
                    return list;
                }
            );

            if (OnSubscriptionAdded != null)
            {
                OnSubscriptionAdded(stock, requestCount);
            }

            if (requestCount == 1)
            {
                if (IsStarted)
                    RegisterQuote(stock);

                if (!_mapSecurities.ContainsKey(stock.Symbol))
                {
                    if (stock is StockSecurity)
                        _mapSecurities.TryAdd(stock.Symbol, (StockSecurity)stock);
                    else if (stock is StockListing)
                        _mapSecurities.TryAdd(stock.Symbol, ((StockListing)stock).Security);
                }
            }
        }

        public void SubscribeLast(IStock stock, IMarketUpdateLast callback)
        {
            long requestCount = 0;

            _mapLastRequests.AddOrUpdate(stock,
                (key) =>
                {
                    requestCount = _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                        (symbol) =>
                        {
                            RequestCounter value = new RequestCounter();
                            value.Increment();
                            return value;
                        },
                        (symbol, value) =>
                        {
                            value.Increment();
                            return value;
                        })
                        .Count;

                    return new List<IMarketUpdateLast>() { callback };
                },
                (key, list) =>
                {
                    if (!list.Contains(callback))
                    {
                        list.Add(callback);

                        requestCount = _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                        (symbol) =>
                        {
                            RequestCounter value = new RequestCounter();
                            value.Increment();
                            return value;
                        },
                        (symbol, value) =>
                        {
                            value.Increment();
                            return value;
                        })
                        .Count;
                    }
                    else { } // don't subscribe twice you twat!
                    return list;
                }
            );

            if (OnSubscriptionAdded != null)
            {
                OnSubscriptionAdded(stock, requestCount);
            }

            if (requestCount == 1)
            {
                if (IsStarted)
                    RegisterQuote(stock);

                if (!_mapSecurities.ContainsKey(stock.Symbol))
                {
                    if (stock is StockSecurity)
                        _mapSecurities.TryAdd(stock.Symbol, (StockSecurity)stock);
                    else if (stock is StockListing)
                        _mapSecurities.TryAdd(stock.Symbol, ((StockListing)stock).Security);
                }
            }
        }

        public void SubscribeOpen(IStock stock, IMarketUpdateOpen callback)
        {
            long requestCount = 0;

            _mapOpenRequests.AddOrUpdate(stock,
                (key) =>
                {
                    requestCount = _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                        (symbol) =>
                        {
                            RequestCounter value = new RequestCounter();
                            value.Increment();
                            return value;
                        },
                        (symbol, value) =>
                        {
                            value.Increment();
                            return value;
                        })
                        .Count;
                                        
                    return new List<IMarketUpdateOpen>() { callback };
                },
                (key, list) =>
                {
                    if (!list.Contains(callback))
                    {
                        list.Add(callback);

                        requestCount = _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                        (symbol) =>
                        {
                            RequestCounter value = new RequestCounter();
                            value.Increment();
                            return value;
                        },
                        (symbol, value) =>
                        {
                            value.Increment();
                            return value;
                        })
                        .Count;
                    }
                    else { } // don't subscribe twice you twat!
                    return list;
                }
            );

            if (OnSubscriptionAdded != null)
            {
                OnSubscriptionAdded(stock, requestCount);
            }

            if (requestCount == 1)
            {
                if (IsStarted)
                    RegisterQuote(stock);

                if (!_mapSecurities.ContainsKey(stock.Symbol))
                {
                    if (stock is StockSecurity)
                        _mapSecurities.TryAdd(stock.Symbol, (StockSecurity)stock);
                    else if (stock is StockListing)
                        _mapSecurities.TryAdd(stock.Symbol, ((StockListing)stock).Security);
                }
            }
        }


        public void SubscribePhase(IStock stock, IMarketUpdatePhase callback)
        {
            long requestCount = 0;

            _mapPhaseRequests.AddOrUpdate(stock,
                (key) =>
                {
                    requestCount = _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                        (symbol) =>
                        {
                            RequestCounter value = new RequestCounter();
                            value.Increment();
                            return value;
                        },
                        (symbol, value) =>
                        {
                            value.Increment();
                            return value;
                        })
                        .Count;

                    return new List<IMarketUpdatePhase>() { callback };
                },
                (key, list) =>
                {
                    if (!list.Contains(callback))
                    {
                        list.Add(callback);

                        requestCount = _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                        (symbol) =>
                        {
                            RequestCounter value = new RequestCounter();
                            value.Increment();
                            return value;
                        },
                        (symbol, value) =>
                        {
                            value.Increment();
                            return value;
                        })
                        .Count;
                    }
                    else { } // don't subscribe twice you twat!
                    return list;
                }
            );

            if (OnSubscriptionAdded != null)
            {
                OnSubscriptionAdded(stock, requestCount);
            }


            if (requestCount == 1)
            {
                if (IsStarted)
                    RegisterQuote(stock);

                if (!_mapSecurities.ContainsKey(stock.Symbol))
                {
                    if (stock is StockSecurity)
                        _mapSecurities.TryAdd(stock.Symbol, (StockSecurity)stock);
                    else if (stock is StockListing)
                        _mapSecurities.TryAdd(stock.Symbol, ((StockListing)stock).Security);
                }
            }
        }



        public void UnsubscribeBid(IStock stock, IMarketUpdateBid callback)
        {
            _mapBidRequests.AddOrUpdate(stock,
                (key) =>
                {
                    // shouldn't ever really come in here
                    _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                        (symbol) =>
                        {
                            if (key is StockSecurity)
                                _mapSecurities.TryAdd(symbol, (StockSecurity)key);
                            else if (key is StockListing)
                                _mapSecurities.TryAdd(symbol, ((StockListing)key).Security);

                            return new RequestCounter();
                        },
                        (symbol, value) =>
                        {
                            return value;
                        });
                    return new List<IMarketUpdateBid>();
                },
                (key, list) =>
                {
                    if (list.Contains(callback))
                    {
                        _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                            (symbol) =>
                            {
                                if (key is StockSecurity)
                                    _mapSecurities.TryAdd(symbol, (StockSecurity)key);
                                else if (key is StockListing)
                                    _mapSecurities.TryAdd(symbol, ((StockListing)key).Security);

                                return new RequestCounter();
                            },
                            (symbol, value) =>
                            {
                                long count = value.Decrement();
                                if (count == 0 && IsStarted)
                                {
                                    DeRegisterQuote(key);
                                }
                                if (OnSubscriptionRemoved != null)
                                    OnSubscriptionRemoved(key, count);
                                return value;
                            });
                        list.Remove(callback);
                    }
                    else { } // no subscription found, don't do that!  
                    return list;
                }
            );
        }

        public void UnsubscribeAsk(IStock stock, IMarketUpdateAsk callback)
        {
            _mapAskRequests.AddOrUpdate(stock,
                (key) =>
                {
                    // shouldn't ever really come in here
                    _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                        (symbol) =>
                        {
                            if (key is StockSecurity)
                                _mapSecurities.TryAdd(symbol, (StockSecurity)key);
                            else if (key is StockListing)
                                _mapSecurities.TryAdd(symbol, ((StockListing)key).Security);

                            return new RequestCounter();
                        },
                        (symbol, value) =>
                        {
                            return value;
                        });
                    return new List<IMarketUpdateAsk>();
                },
                (key, list) =>
                {
                    if (list.Contains(callback))
                    {
                        _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                            (symbol) =>
                            {
                                if (key is StockSecurity)
                                    _mapSecurities.TryAdd(symbol, (StockSecurity)key);
                                else if (key is StockListing)
                                    _mapSecurities.TryAdd(symbol, ((StockListing)key).Security);

                                return new RequestCounter();
                            },
                            (symbol, value) =>
                            {
                                long count = value.Decrement();
                                if (count == 0 && IsStarted)
                                {
                                    DeRegisterQuote(key);
                                }
                                if (OnSubscriptionRemoved != null)
                                    OnSubscriptionRemoved(key, count);
                                return value;
                            });
                        list.Remove(callback);
                    }
                    else { } // no subscription found, don't do that!  
                    return list;
                }
            );
        }

        public void UnsubscribeLast(IStock stock, IMarketUpdateLast callback)
        {
            _mapLastRequests.AddOrUpdate(stock,
                (key) =>
                {
                    // shouldn't ever really come in here
                    _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                        (symbol) =>
                        {
                            if (key is StockSecurity)
                                _mapSecurities.TryAdd(symbol, (StockSecurity)key);
                            else if (key is StockListing)
                                _mapSecurities.TryAdd(symbol, ((StockListing)key).Security);

                            return new RequestCounter();
                        },
                        (symbol, value) =>
                        {
                            return value;
                        });
                    return new List<IMarketUpdateLast>();
                },
                (key, list) =>
                {
                    if (list.Contains(callback))
                    {
                        _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                            (symbol) =>
                            {
                                if (key is StockSecurity)
                                    _mapSecurities.TryAdd(symbol, (StockSecurity)key);
                                else if (key is StockListing)
                                    _mapSecurities.TryAdd(symbol, ((StockListing)key).Security);

                                return new RequestCounter();
                            },
                            (symbol, value) =>
                            {
                                long count = value.Decrement();
                                if (count == 0 && IsStarted)
                                {
                                    DeRegisterQuote(key);
                                }
                                if (OnSubscriptionRemoved != null)
                                    OnSubscriptionRemoved(key, count);
                                return value;
                            });
                        list.Remove(callback);
                    }
                    else { } // no subscription found, don't do that!  
                    return list;
                }
            );
        }

        public void UnsubscribeOpen(IStock stock, IMarketUpdateOpen callback)
        {
            _mapOpenRequests.AddOrUpdate(stock,
                (key) =>
                {
                    // shouldn't ever really come in here
                    _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                        (symbol) =>
                        {
                            if (key is StockSecurity)
                                _mapSecurities.TryAdd(symbol, (StockSecurity)key);
                            else if (key is StockListing)
                                _mapSecurities.TryAdd(symbol, ((StockListing)key).Security);

                            return new RequestCounter();
                        },
                        (symbol, value) =>
                        {
                            return value;
                        });
                    return new List<IMarketUpdateOpen>();
                },
                (key, list) =>
                {
                    if (list.Contains(callback))
                    {
                        _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                            (symbol) =>
                            {
                                if (key is StockSecurity)
                                    _mapSecurities.TryAdd(symbol, (StockSecurity)key);
                                else if (key is StockListing)
                                    _mapSecurities.TryAdd(symbol, ((StockListing)key).Security);

                                return new RequestCounter();
                            },
                            (symbol, value) =>
                            {
                                long count = value.Decrement();
                                if (count == 0 && IsStarted)
                                {
                                    DeRegisterQuote(key);
                                }
                                if (OnSubscriptionRemoved != null)
                                    OnSubscriptionRemoved(key, count);
                                return value;
                            });
                        list.Remove(callback);
                    }
                    else { } // no subscription found, don't do that!  
                    return list;
                }
            );
        }



        public void UnsubscribePhase(IStock stock, IMarketUpdatePhase callback)
        {
            _mapPhaseRequests.AddOrUpdate(stock,
                (key) =>
                {
                    // shouldn't ever really come in here
                    _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                        (symbol) =>
                        {
                            if (key is StockSecurity)
                                _mapSecurities.TryAdd(symbol, (StockSecurity)key);
                            else if (key is StockListing)
                                _mapSecurities.TryAdd(symbol, ((StockListing)key).Security);

                            return new RequestCounter();
                        },
                        (symbol, value) =>
                        {
                            return value;
                        });
                    return new List<IMarketUpdatePhase>();
                },
                (key, list) =>
                {
                    if (list.Contains(callback))
                    {
                        _mapSecuritiesCount.AddOrUpdate(key.Symbol,
                            (symbol) =>
                            {
                                if (key is StockSecurity)
                                    _mapSecurities.TryAdd(symbol, (StockSecurity)key);
                                else if (key is StockListing)
                                    _mapSecurities.TryAdd(symbol, ((StockListing)key).Security);

                                return new RequestCounter();
                            },
                            (symbol, value) =>
                            {
                                long count = value.Decrement();
                                if (count == 0 && IsStarted)
                                {
                                    DeRegisterQuote(key);
                                }
                                if (OnSubscriptionRemoved != null)
                                    OnSubscriptionRemoved(key, count);
                                return value;
                            });
                        list.Remove(callback);
                    }
                    else { } // no subscription found, don't do that!  
                    return list;
                }
            );
        }



        #endregion


        public void SubscribeRawUpdates(IStock stock)
        {
            long requestCount = 0;

            requestCount = _mapRawRequests.AddOrUpdate(stock,
                (key) =>
                {
                    return 1;
                },
                (key, count) =>
                {
                    return ++count;
                }
            );

            if (OnSubscriptionAdded != null)
            {
                OnSubscriptionAdded(stock, requestCount);
            }

            if (requestCount == 1)
            {
                if (IsStarted)
                    RegisterQuote(stock);

                if (!_mapSecurities.ContainsKey(stock.Symbol))
                {
                    if (stock is StockSecurity)
                        _mapSecurities.TryAdd(stock.Symbol, (StockSecurity)stock);
                    else if (stock is StockListing)
                        _mapSecurities.TryAdd(stock.Symbol, ((StockListing)stock).Security);
                }
            }
        }

        public void UnsubscribeRawUpdates(IStock stock)
        {
            _mapRawRequests.AddOrUpdate(stock,
                (key) =>
                {
                    // shouldn't ever really come in here
                    return 0;
                },
                (key, count) =>
                {
                    if (--count == 0 && IsStarted)
                    {
                        DeRegisterQuote(key);
                    }

                    if (OnSubscriptionRemoved != null)
                        OnSubscriptionRemoved(key, count);

                    return count;
                }
            );
        }

        private void RegisterQuote(IStock stock)
        {
            if (stock is StockSecurity)
            {
                base.RegisterQuote(stock.Symbol, "*");
            }
            else if (stock is StockListing)
            {
                base.RegisterQuote(stock.Symbol, "*");
            }
        }

        private void DeRegisterQuote(IStock stock)
        {
            base.DeRegisterQuote(stock.Symbol, "*");
        }

        #endregion


    }

}

